#include "../include/Connection.h"
#include "../include/EventLoop.h"

Connection::Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
    : _conn_id(conn_id),
      _sockfd(sockfd),
      _enable_inactive_release(false),
      _statu(ConnStatu::CONNECTING),
      _socket(sockfd),
      _channel(Channel(loop, sockfd)),
      _loop(loop)
{
    // channel的读事件监控不能再这里启用，定时任务会出问题
    _channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));
    _channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
    _channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
    _channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));
    _channel.SetRandomCallBack(std::bind(&Connection::HandleRandom, this));
}

Connection::~Connection() { DBG_LOG("release connection %p", this); }

int Connection::getFd() { return _sockfd; }

uint64_t Connection::getId() { return _conn_id; }

bool Connection::getConnected() { return _statu == ConnStatu::CONNECTED; }

void Connection::SetContext(const Any &any) { _context = any; }

Any *Connection::getContext() { return &_context; }

void Connection::Established() { _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this)); }

void Connection::Send(const char *data, uint32_t len) { _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len)); }

void Connection::ShutDown() { _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this)); }

void Connection::EnableInactiveRelease(uint64_t sec) { _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec)); }

void Connection::DisableInactiveRelease() { _loop->RunInLoop(std::bind(&Connection::DisableInactiveReleaseInLoop, this)); }

// 必须保证线程安全，可能会出现协议切换没有执行的的情况，这样新的数据使用了旧的协议处理，肯定是不对的
// 所以必须在当前线程内立即执行
void Connection::Upgrade(const Any &any,
                         const ConnectedCallBack &conn_cb,
                         const RandomEventCallBack &ra_cb,
                         const ClosedCallBack &cl_cb,
                         const MessageCallBack &mes_cb)
{
    _loop->AssertInLoop();
    _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, any, conn_cb, ra_cb, cl_cb, mes_cb));
}

void Connection::SetConnectedCallBack(const ConnectedCallBack &conn_cb) { _conn_cb = conn_cb; }

void Connection::Release() { _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this)); }
void Connection::SetMessageCallBack(const MessageCallBack &mes_cb) { _mess_cb = mes_cb; }

void Connection::SetRandomEventCallBack(const RandomEventCallBack &ra_cb) { _rand_ev_cb = ra_cb; }

void Connection::SetClosedCallBack(const ClosedCallBack &cl_cb) { _close_cb = cl_cb; }

void Connection::SetServerClosedCallBack(const ClosedCallBack &ser_cl_cb) { _server_close_cb = ser_cl_cb; }

void Connection::HandleRead()
{
    // ET 模式下必须循环读取所有数据，直到 EAGAIN
    char buffer[65536];
    while (true)
    {
        ssize_t ret = _socket.NonBlockRecv(buffer, sizeof(buffer) - 1);

        if (ret > 0)
        {
            _in_buffer.WriteAndPush(buffer, ret);
            // 继续循环读取更多数据
        }
        else if (ret == 0)
        {
            // ret == 0 indicates peer shutdown
            // 先处理已读取的数据
            if (_in_buffer.ReadableSpace() > 0 && _mess_cb)
            {
                _mess_cb(shared_from_this(), &_in_buffer);
            }
            return ShutDownInLoop();
        }
        else if (ret == SOCKET_EAGAIN)
        {
            // SOCKET_EAGAIN means no more data available
            // 处理已读取的数据，然后返回等待下次事件
            break;
        }
        else
        {
            // ret < 0 (and not SOCKET_EAGAIN) indicates a real error
            if (_in_buffer.ReadableSpace() > 0 && _mess_cb)
            {
                _mess_cb(shared_from_this(), &_in_buffer);
            }
            return ShutDownInLoop();
        }
    }

    // 循环结束后，处理所有读取到的数据
    if (_in_buffer.ReadableSpace() > 0 && _mess_cb)
    {
        _mess_cb(shared_from_this(), &_in_buffer);
    }
}

void Connection::HandleWrite()
{
    // ET 模式下循环发送所有数据，直到缓冲区为空或 EAGAIN
    while (_out_buffer.ReadableSpace() > 0)
    {
        ssize_t ret = _socket.NonBlockSend(_out_buffer.getReadOffset(), _out_buffer.ReadableSpace());

        if (ret < 0 && ret != SOCKET_EAGAIN)
        {
            // send出错说明连接不可靠或者对端关闭了,接着保持连接无意义或者代价很高，直接关闭连接
            if (_in_buffer.ReadableSpace() > 0 && _mess_cb)
                _mess_cb(shared_from_this(), &_in_buffer);

            return ReleaseInLoop();
        }

        if (ret == SOCKET_EAGAIN)
        {
            // Would block, wait for next write event
            return;
        }

        _out_buffer.MoveReadOffset(ret);
    }

    // 所有数据发送完毕
    if (_out_buffer.ReadableSpace() == 0)
    {
        // 没有数据待发送，就关闭写事件监控
        _channel.DisableWrite();
        // 如果是待关闭状态且没有数据待发送，就关闭连接
        if (_statu == ConnStatu::DISCONNECTING)
            return ReleaseInLoop();
    }
}

void Connection::HandleClose()
{
    // 连接挂断后，套接字就什么也处理不了，处理完数据，关闭即可
    if (_in_buffer.ReadableSpace() > 0)
        _mess_cb(shared_from_this(), &_in_buffer);

    return ReleaseInLoop();
}

void Connection::HandleError() { HandleClose(); }

void Connection::HandleRandom()
{
    // 1.刷新活跃度
    if (_enable_inactive_release)
        _loop->TimerRefersh(_conn_id);
    // 2.调用组件使用者的任意事件回调
    if (_rand_ev_cb)
        _rand_ev_cb(shared_from_this());
}

// 获取连接之后，所处的状态下要进行各种设置(启动读监控，调用回调函数)
void Connection::EstablishedInLoop()
{
    // 1.修改连接状态,必须是连接中的状态
    assert(_statu == ConnStatu::CONNECTING);
    _statu = ConnStatu::CONNECTED;
    // 2.启动读事件监控
    _channel.EnableRead();
    // 3.调用回调函数
    if (_conn_cb)
        _conn_cb(shared_from_this());
}

void Connection::ReleaseInLoop()
{
    // 1.修改连接状态
    _statu = ConnStatu::DISCONNECTED;
    // 2.移除连接的事件监控
    _channel.Remove();
    // 3.关闭文件描述符
    ::close(_sockfd);
    // 4.如果定时器队列还有定时销毁任务，就取消
    if (_loop->HasTimer(_conn_id))
        DisableInactiveReleaseInLoop();
    // 5. 调用关闭回调函数,先调用使用者的组件，防止先调用服务器的会导致使用者的连接断开
    if (_close_cb)
        _close_cb(shared_from_this());
    if (_server_close_cb)
        _server_close_cb(shared_from_this());
}

void Connection::SendInLoop(const char *data, uint32_t len)
{
    // 确认是连接状态
    if (_statu == ConnStatu::DISCONNECTED)
        return;
    _out_buffer.WriteAndPush(data, len);

    if (!_channel.WriteAble())
        _channel.EnableWrite();
}

void Connection::ShutDownInLoop()
{
    _statu = ConnStatu::DISCONNECTING;
    if (_in_buffer.ReadableSpace() > 0)
        if (_mess_cb)
            _mess_cb(shared_from_this(), &_in_buffer);

    if (_out_buffer.ReadableSpace() > 0)
        if (!_channel.ReadAble())
            _channel.EnableRead();
    // 没有待发送数据，关闭
    if (_out_buffer.ReadableSpace() == 0)
        ReleaseInLoop();
}

void Connection::EnableInactiveReleaseInLoop(uint64_t sec)
{
    _enable_inactive_release = true;
    // 如果定时销毁任务存在，刷新一下就行，不存在在添加
    if (_loop->HasTimer(_conn_id))
        return _loop->TimerRefersh(_conn_id);
    // 调用释放连接的函数
    _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::ReleaseInLoop, this));
}

void Connection::DisableInactiveReleaseInLoop()
{
    _enable_inactive_release = false;
    if (_loop->HasTimer(_conn_id))
        _loop->TimerCancel(_conn_id);
}

void Connection::UpgradeInLoop(const Any &any,
                               const ConnectedCallBack &conn_cb,
                               const RandomEventCallBack &ra_cb,
                               const ClosedCallBack &cl_cb,
                               const MessageCallBack &mes_cb)
{
    SetContext(any);
    SetClosedCallBack(cl_cb);
    SetConnectedCallBack(conn_cb);
    SetMessageCallBack(mes_cb);
    SetRandomEventCallBack(ra_cb);
}
